# Python3_DataProcMultiThreading.py
# Create By GF 2023-11-27 17:24

# ----------------------------------------------------------------------------------------------------

# 数据处理 - 多线程 (Data Processing - Multi Threading).

# ----------------------------------------------------------------------------------------------------

import threading

def Pandas_Compute_Field_Segmented_Index(DataFrame, Number_Of_Segments:int):

    Max_Index:int = DataFrame.index.max()
    # ----------------------------------------------
    Number_Of_Rows_Per_Segment:int = int(Max_Index / Number_Of_Segments)
    # ----------------------------------------------
    Index_List:list = []
    # ----------------------------------------------
    Counter:int = 1
    Bgn_Idx:int = 0
    while Counter <= Number_Of_Segments:
        if (Counter == 1)                                  : Index_List.append((Bgn_Idx, Bgn_Idx + Number_Of_Rows_Per_Segment - 1))
        if (1 < Counter) and (Counter < Number_Of_Segments): Index_List.append((Bgn_Idx, Bgn_Idx + Number_Of_Rows_Per_Segment - 1))
        if (Counter == Number_Of_Segments)                 : Index_List.append((Bgn_Idx, Max_Index))
        # ------------------------------------------
        Bgn_Idx = Bgn_Idx + Number_Of_Rows_Per_Segment
        Counter = Counter + 1
    # ----------------------------------------------
    return Index_List

def Pandas_Compute_Field_Sales_Package(DataFrame, BgnIdx:int, EndIdx:int):

    DataFrame["套餐标记"] = None
    # ----------------------------------------------
    for Idx in range(BgnIdx, (EndIdx + 1)):
        Order_Number = DataFrame.loc[Idx, "订单号"]
        # ------------------------------------------
        Filtered_DataFrame   = DataFrame[DataFrame["订单号"] == Order_Number]
        Filtered_Rows_Number = Filtered_DataFrame["订单号"].count()
        Filtered_Index_List  = Filtered_DataFrame.index
        # ------------------------------------------
        if (Filtered_Rows_Number >= 2) and (DataFrame.loc[Idx, "套餐标记"] == None):
            DataFrame.loc[Filtered_Index_List, "套餐标记"] = "Package(%d)" % Order_Number
        else:
            pass
    # ----------------------------------------------
    # Function End.

def Pandas_Compute_Field_Calling_Multi_Threading(Args_1):

    SegIdx = Pandas_Compute_Field_Segmented_Index(Args_1, 5)
    # ----------------------------------------------
    Thread_Pool = [threading.Thread(target=Pandas_Compute_Field_Sales_Package, args=[Args_1, SegIdx[0][0], SegIdx[0][1]], name='Th_1'),
                   threading.Thread(target=Pandas_Compute_Field_Sales_Package, args=[Args_1, SegIdx[1][0], SegIdx[1][1]], name='Th_2'),
                   threading.Thread(target=Pandas_Compute_Field_Sales_Package, args=[Args_1, SegIdx[2][0], SegIdx[2][1]], name='Th_3'),
                   threading.Thread(target=Pandas_Compute_Field_Sales_Package, args=[Args_1, SegIdx[3][0], SegIdx[3][1]], name='Th_4'),
                   threading.Thread(target=Pandas_Compute_Field_Sales_Package, args=[Args_1, SegIdx[4][0], SegIdx[4][1]], name='Th_5')]
    # ----------------------------------------------
    for Th in Thread_Pool:
        # 开始 Start: 在创建了线程类之后，你需要通过 start() 函数来开始运行你的目标函数。
        #             这里通过对线程池列表的循环来一一启动其目标函数。
        # 非阻塞 Non-Blocking: 线程类的 start() 函数是一个非阻塞函数，意思是当目标函数启动了之后，程序就返回了，而不是等到目标函数结束再返回。
        #                      所以这里可以实现对线程池的线程同时启动，而不是等待前一个结束再开始下一个。
        Th.start()
    # ----------------------------------------------
    for Th in Thread_Pool:
        # 加入 Join: 第二次循环线程池列表的时候，调用了线程类的 join() 函数。
        #            join() 函数会等待目标函数运行结束之后返回，是一个阻塞 (Blocking) 函数。
        #            目前教程介绍的都是阻塞型函数，在之后的教程中讲到 异步(Asynchronous) 的时候会了解如何写非阻塞函数。
        Th.join()
    # ----------------------------------------------
    # Function End.